package com.shujia.mr.wc1;

/*
    通过我们之前对mapreduce的描述来说，有两个阶段：分别是map阶段和reduce阶段
    并且map任务如何处理以及reduce任务如何处理数据，这两个部分应该是开发者可以自定义的
    应该是两个方法，所以一个完整的mapreduce程序需要写三个类：
    1）Map类，其中要描述map任务要做的事情
    2）Reduce类，其中要描述reduce任务要做的事情
    3）测试类，包含main方法，用于设置并提交的整个mapreduce作业的、

    整个称为mapreduce作业，包含若干个map任务线程和reduce任务线程，这个作用是通过测试类中的main方法去启动的。

    测试类中要对整个mr作业进行设置，例如设置作业的名字，reduce的个数，切片的大小，map任务的输入输出类型
    reduce的输入输出类型，map类是谁，reduce类是谁，提交作业到yarn上申请资源开始执行。

 */

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/*
    运行方式：
    先启动hadoop集群 start-all.sh
        1、打包：
            1）借助idea打包
            2）借助maven组件打包【推荐】
        2、将jar包上传到linux服务器中
        3、上传数据到hdfs上，用户是root用户
        4、使用命令运行程序
            hadoop jar hadoop-1.0-SNAPSHOT.jar com.shujia.mr.wc1.WordCountDemo1 /bigdata33/data/words.txt /bigdata33/output/out1
 */

public class WordCountDemo1 {
    public static void main(String[] args) {
        try {
            //创建hadoop配置文件对象
            Configuration conf = new Configuration();

            //创建MapReduce作业对象
            Job job = Job.getInstance(conf);
            //设置mr作业的名字
            job.setJobName("33期单词统计mr实现案例1");
            //设置reduce的个数
//            job.setNumReduceTasks(2);

            //预聚合，提前在map端聚合一次，聚合逻辑这里和reduce执行逻辑是一样
            job.setCombinerClass(WordCountReducer.class);

            //设置mr作业的主类
            job.setJarByClass(WordCountDemo1.class);

            //设置map类
            job.setMapperClass(WordCountMapper.class);
            //设置reduce类
            job.setReducerClass(WordCountReducer.class);

            //设置map任务的输出数据类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);

            //设置reduce任务的输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);

            // 设置读取hdfs文件的路径
//            FileInputFormat.addInputPath(); //添加某一个目录
//            FileInputFormat.setInputPaths(); // 可以添加多个目录
//            FileInputFormat.setInputPaths(job, new Path("/xxx/xxx/xx"));
            FileInputFormat.setInputPaths(job, new Path(args[0]));

            //设置输出的hdfs目录路径
            FileOutputFormat.setOutputPath(job,new Path(args[1]));

            //提交作业到yarn中运行
            boolean b = job.waitForCompletion(true);

            if(b){
                System.out.println("========33期单词统计mr实现案例1执行成功！！！！========");
                System.out.println("输出结果目标在hdfs路径："+args[1]);
            }else {
                System.out.println("-------------- 执行失败！！ -----------------");
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
